package com.allen.flink.book;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * 功能:并行度为1的发送器，用来向kafka发送数据
 *
 * @date: 2020-03-22 23:31
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
public class MyNoParallelSource implements SourceFunction<String> {
    
    //private long count = 1L;
    private boolean isRunning = true;
    
    /**
     * 主要的方法
     * 启动一个source
     * 大部分情况下，都需要在这个run方法中实现一个循环，这样就可以循环产生数据了
     *
     * @param ctx
     *
     * @throws Exception
     */
    
    public void run(final SourceContext<String> ctx) throws Exception {
        while (isRunning) {
            //图书的排行榜
            List<String> books = new ArrayList<>();
            books.add("Python从入门到放弃");//10
            books.add("Java从入门到放弃");//8
            books.add("Php从入门到放弃");//5
            books.add("C++从入门到放弃");//3
            books.add("Scala从入门到放弃");//0-4
            int i = new Random().nextInt(5);
            ctx.collect(books.get(i));
            
            //每1秒产生一条数据
            Thread.sleep(1000);
        }
        
    }
    
    //取消一个cancel的时候会调用的方法
    public void cancel() {
        isRunning = false;
    }
}
